package com.storm.exam3.q2;

import org.apache.storm.trident.operation.BaseAggregator;
import org.apache.storm.trident.operation.TridentCollector;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Values;

/**
 * #Author :Sino
 * #Date   :2022/11/23 15:02
 * #Describe:
 */
public class Averager extends BaseAggregator<Double> {
    @Override
    public Double init(Object batchId, TridentCollector collector) {
        return 0.;
    }

    @Override
    public void aggregate(Double val, TridentTuple tuple, TridentCollector collector) {
        ++val;
        tuple.getDoubleByField("score");
    }

    @Override
    public void complete(Double val, TridentCollector collector) {
        collector.emit(new Values("score"));
    }
}
